package SparkExercises;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFunction;
import scala.Serializable;
import scala.Tuple2;

import java.util.List;

public class ZuoYe2 {
    public static void main(String[] args) {
        /**
         * 已知有一份新能源车辆行程数据trip.csv（见附件），
         * 它的第一列表示车辆的车架号（每辆车都是唯一的），
         * 第二列表示车型，第三列表示行程日期，第四列表示当天行驶的总里程（单位是公里），
         * 第五列表示当天行驶的总时长（单位是分钟）。
         * 请写一个spark程序，计算出每辆车行驶总里程最大的日期，并把对应的记录输出到控制台，如下图所示：
         */
        SparkConf sparkConf = new SparkConf().setAppName("ZuoYe2").setMaster("local");
        JavaSparkContext sc = new JavaSparkContext(sparkConf);

        JavaRDD<String> rdd1 = sc.textFile("./data/trip2.csv");
        JavaRDD<Car> rdd2 = rdd1.map(new Function<String, Car>() {
            @Override
            public Car call(String s) throws Exception {
                String[] split = s.split(",");
                Car car = new Car(split[0], split[1], split[2], Integer.parseInt(split[3]), Integer.parseInt(split[4]));
                return car;
            }
        });
        JavaPairRDD<String, Car> rdd3 = rdd2.mapToPair(new PairFunction<Car, String, Car>() {
            @Override
            public Tuple2<String, Car> call(Car car) throws Exception {
                return new Tuple2<>(car.getVin(), car);
            }
        });

        JavaPairRDD<String, Car> rdd4 = rdd3.reduceByKey(new Function2<Car, Car, Car>() {
            @Override
            public Car call(Car car, Car car2) throws Exception {
                if (car.getMinutes() > car2.getMinutes()) {
                    return car;
                } else {
                    return car2;
                }
            }
        });
        JavaRDD<Car> rdd5 = rdd4.values();
        List<Car> collect = rdd5.collect();
        for (Car c :collect){
            System.out.println(c);
        }
//        List<Tuple2<String, Car>> collect = rdd4.collect();
//        for (Tuple2<String, Car> t: collect){
//            System.out.println(t);
//        }


    }
}
class Car implements Serializable {
    private String vin;
    private String vehicle_type;
    private String date;
    private int mileages;
    private int minutes;

    public Car(String vin, String vehicle_type, String date, int mileages, int minutes) {
        this.vin = vin;
        this.vehicle_type = vehicle_type;
        this.date = date;
        this.mileages = mileages;
        this.minutes = minutes;
    }

    @Override
    public String toString() {
        return "Car{" +
                "vin='" + vin + '\'' +
                ", vehicle_type='" + vehicle_type + '\'' +
                ", date='" + date + '\'' +
                ", mileages=" + mileages +
                ", minutes=" + minutes +
                '}';
    }

    public String getVin() {
        return vin;
    }

    public void setVin(String vin) {
        this.vin = vin;
    }

    public String getVehicle_type() {
        return vehicle_type;
    }

    public void setVehicle_type(String vehicle_type) {
        this.vehicle_type = vehicle_type;
    }

    public String getDate() {
        return date;
    }

    public void setDate(String date) {
        this.date = date;
    }

    public int getMileages() {
        return mileages;
    }

    public void setMileages(int mileages) {
        this.mileages = mileages;
    }

    public int getMinutes() {
        return minutes;
    }

    public void setMinutes(int minutes) {
        this.minutes = minutes;
    }
}

